package cn.azzhu.day02
import org.apache.flink.api.common.functions.FilterFunction
import org.apache.flink.streaming.api.scala._
/**
 * @author azzhu
 * @create 2020-09-20 23:57:32
 */
object FilterExample {
  def main(args: Array[String]): Unit = {
    //todo 1.获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //todo 2.数据源
    val stream = env.addSource(new SensorSource)

    stream.filter(r => r.id.equals("sensor_1")).print()
    stream.filter(new FilterFunction[SensorReading] {
      override def filter(t: SensorReading): Boolean = t.id.equals("sensor_1")
    })

    stream.filter(new MyFilterFunction)

    env.execute("FilterExample")
  }

  //filter算子输入和输出的类型是一样的，所以只有一个泛型SensorReading
  class MyFilterFunction extends FilterFunction[SensorReading] {
    override def filter(t: SensorReading): Boolean = t.id.equals("sensor_1")
  }
}
